// Copyright (c) 2021 Presto Labs Pte. Ltd.
// Author: gekim

#include "coin2/exchange/bybit/feed_v1/processor.h"

#include <string>
#include <utility>

#include "coin2/base/conversion.h"
#include "coin2/exchange/base/api_util/json_schema.h"
#include "coin2/exchange/bybit/feed_v1/update.h"
#include "coin2/exchange/util/json_util.h"

namespace coin2::exchange::bybit_spot::feed_v1::impl {
void BybitFeedProcessor::OnTopicRecord(
    const base::executor::TopicRecordPacket& packet) {
  upd_mgr_.onBeginPacket(packet);

  rapidjson::Document document;
  document.Parse<rapidjson::kParseNoFlags>(packet.payload.data());
  if (document.HasParseError()) {
    throw std::runtime_error(std::string("JSON parse error: ") +
                             GetParseError_En(document.GetParseError()));
  }

  if (parser_config_.validate_schema() && schema_) {
    schema_->AcceptOrThrow(document);
  }

  if (document.HasMember("pong")) {
    upd_mgr_.onEndPacket(packet);
    return;
  }

  if (!document.HasMember("url") &&
      (!document.HasMember("topic") || !document.HasMember("symbol"))) {
    LOG(INFO) << "[Bybit Feed Processor] meta msg for " << packet.topic_name
              << ": " << packet.payload.data();
    upd_mgr_.onEndPacket(packet);
    return;
  }

  std::string channel;
  std::string native_symbol;
  std::string url;
  std::map<std::string, std::string> parameters;
  if (document.HasMember("url")) {
    url = document["url"].GetString();
    std::vector<std::string> splited = Split(url, "?");
    CHECK_EQ(splited.size(), 2);
    if (StringEndsWith(splited[0], "klines")) {
      channel = "kline_rest";
      parameters = base::feed::FeedParsingProcessor::GetRequestParameters(url);
      native_symbol = parameters["symbol"];
      if (StringStartsWith(native_symbol, "301.")) {
        native_symbol = native_symbol.substr(4);
      }
    }
  } else {
    channel = document["topic"].GetString();
    native_symbol = document["symbol"].GetString();
  }

  if (channel == "depth") {
    // fastfeed back compatible, bybit spot only subscribe 1 book
    if (!CheckChannelFilter("book")) {
      upd_mgr_.onEndPacket(packet);
      return;
    }
  } else {
    if (!CheckChannelFilter(channel)) {
      upd_mgr_.onEndPacket(packet);
      return;
    }
  }

  if (!upd_mgr_.is_symbol_registered(native_symbol)) {
    upd_mgr_.onEndPacket(packet);
    return;
  }

  if (channel == "depth" || channel == "diffDepth") {
    auto* book = upd_mgr_.GetBookMutable(native_symbol);
    bool is_diff = (channel == "diffDepth");
    if (parser_config_.has_sampling_frequency_bps()) {
      bool generated = BybitSpotFeedUpdateParser::ParseBook(book, document, is_diff);
      if (generated) {
        bool sampled = book->SampleByFrequency(
            book->Bid0(), book->Ask0(),
            parser_config_.sampling_frequency_bps());
        if (parser_config_.has_trade_sampling_frequency_bps()) {
          auto* trade = upd_mgr_.GetTradeMutable();
          auto* slim_trade = upd_mgr_.GetSlimTradeMutable(native_symbol);
          int64_t timespan = packet.timestamp - slim_trade->LastPublishedTimestamp();
          if (sampled ||
              timespan > parser_config_.trade_sampling_time_length()) {
            if (slim_trade->PopulateTradeAndClear(trade)) {
              upd_mgr_.PublishTrade(native_symbol, false);
            }
          }
        }
        if (sampled) {
          upd_mgr_.PublishBook();
        }
      }
    } else {
      if (BybitSpotFeedUpdateParser::ParseBook(book, document, is_diff)) {
        upd_mgr_.PublishBook();
      }
    }
  } else if (channel == "trade") {
    if (parser_config_.has_trade_sampling_frequency_bps()) {
      auto* slim_trade = upd_mgr_.GetSlimTradeMutable(native_symbol);
      for (const auto& trade_json : document["data"].GetArray()) {
        Trade _trade{};
        BybitSpotFeedUpdateParser::ParseTrade(trade_json, &_trade);
        bool aggregated = slim_trade->TryAggregateSlimTrade(
            packet.timestamp,
            _trade.price,
            _trade.fill_qty,
            _trade.side,
            _trade.timestamp,
            parser_config_.trade_sampling_frequency_bps());
        int64_t timespan = packet.timestamp - slim_trade->LastPublishedTimestamp();
        if (!aggregated ||
            timespan > parser_config_.trade_sampling_time_length()) {
          auto* trade = upd_mgr_.GetTradeMutable();
          CHECK(slim_trade->PopulateTradeAndClear(trade));
          upd_mgr_.PublishTrade(native_symbol, false);
        }
      }
    } else {
      Trade* agg_trade = upd_mgr_.GetTradeMutable();
      agg_trade->Clear();
      Trade trade{};
      for (auto& trade_json : document["data"].GetArray()) {
        BybitSpotFeedUpdateParser::ParseTrade(trade_json, &trade);
        if (TryAggregateTrades(agg_trade, trade)) {
          continue;
        }
        if (agg_trade->fill_qty > 0) {
          upd_mgr_.PublishTrade(native_symbol, true);
        }
        *agg_trade = trade;
      }
      if (agg_trade->fill_qty > 0) {
        upd_mgr_.PublishTrade(native_symbol, false);
      }
    }
  } else if (channel == "kline_rest") {
    auto interval = BybitSpotFeedUpdateParser::ParseKlineInterval(parameters["interval"]);
    auto* klines = upd_mgr_.GetKlineMutable(native_symbol, interval);
    if (klines != nullptr) {
      auto data = document["data"]["result"].GetArray();
      std::optional<int64_t> first_open_timestamp;
      if (data.Size() > 0) {
        first_open_timestamp = data[0]["t"].GetInt64() * 1'000'000LL;
      }

      if (klines->Size() == 0) {
        klines->SnapshotClear();
      } else if (first_open_timestamp &&
                 first_open_timestamp < klines->RBegin()->open_timestamp) {
        klines->SnapshotClear();
      } else {
        klines->UpdateClear();
      }
      bool to_publish = false;
      for (const auto& kline_data : data) {
        bool freezed = BybitSpotFeedUpdateParser::ParseKlineRest(kline_data,
            klines, interval, packet.timestamp);
        to_publish = to_publish || freezed;
      }
      if (to_publish) upd_mgr_.PublishKline(native_symbol);
    }
  } else {
    THROW() << "[Bybit Feed Processor] unkown topic: " << packet.payload;
  }

  upd_mgr_.onEndPacket(packet);
}  // namespace coin2::exchange::bybit::feed_v1::impl
}  // namespace coin2::exchange::bybit::feed_v1::impl
